package com.atguigu.flow;

import lombok.extern.slf4j.Slf4j;

import java.util.concurrent.Flow;
import java.util.concurrent.SubmissionPublisher;

/**
 * @author Jungle
 * @create 2024-01-08 22:17
 */
@Slf4j
public class FlowDemo {
   static class  MyProcessor extends SubmissionPublisher<String> implements Flow.Processor<String,String>{
        private Flow.Subscription subscription;  //保存绑定关系

        @Override
        public void onSubscribe(Flow.Subscription subscription) {
            System.out.println("processor订阅绑定完成");
            this.subscription = subscription;
            subscription.request(1);  //找上游要一个数据
        }

        @Override  //数据到达，触发这个回调
        public void onNext(String item) {
            System.out.println("processor拿到数据： "+item);
            //再加工
            item += "haha";
            submit(item);  //把我加工的数据发出去
            subscription.request(1);  //再要新数据发出去
        }

        @Override
        public void onError(Throwable throwable) {

        }

        @Override
        public void onComplete() {

        }
    }
    /**
     * 1、Publisher 发布者
     * 2、Subscriber 订阅者
     * 3、Subscription 订阅关系
     * 4、Processor 处理器
     *
     * @param args
     */
    // 发布订阅模型 ：观察者模式
    public static void main(String[] args) {
        // 1、定义一份发布者；发布数据
        SubmissionPublisher<String> publisher = new SubmissionPublisher<>();

        //2、 定义中间操作，给每个元素加哈哈前缀
        MyProcessor myProcessor = new MyProcessor();
        MyProcessor myProcessor1 = new MyProcessor();
        MyProcessor myProcessor2 = new MyProcessor();

        // 3、定义一个订阅者，订阅者感兴趣发布者的数据
        Flow.Subscriber<Object> subscriber = new Flow.Subscriber<>() {
            private Flow.Subscription subscription;

            @Override  //onxxx  ：在xxx事件发生时候，执行这个回调
            public void onSubscribe(Flow.Subscription subscription) {
                System.out.println(Thread.currentThread().getName() + "订阅开始了：" + subscription);
                this.subscription = subscription;
                // 从上游请求一个数据
                subscription.request(1);
            }

            @Override  //onxxx  ：在下一个元素到达时候，执行这个回调==》接收到了新数据
            public void onNext(Object item) {
                System.out.println(Thread.currentThread().getName() + "\t 订阅者，接收到数据" + item);
                // subscription.request(1);
                if (item.equals("p-7")) {
                    subscription.cancel(); //取消订阅
                }else {
                    subscription.request(1);
                }
            }

            @Override //在错误发生时
            public void onError(Throwable throwable) {
                System.out.println(Thread.currentThread().getName() + "\t 订阅者接收到错误信号：" + throwable);
            }

            @Override  //在完成时候
            public void onComplete() {
                System.out.println(Thread.currentThread().getName() + "\t 订阅者接收到完成信号：");
            }
        };

        //4、绑定发布者和订阅者
        publisher.subscribe(myProcessor);  //此时处理器相当于订阅者
        myProcessor.subscribe(myProcessor1); //此时处理器相当于发布者
        myProcessor1.subscribe(myProcessor2);
        myProcessor2.subscribe(subscriber);  //链表关系绑定出责任链
        //绑定操作;就是发布者，记住了所有订阅者都有谁，有数据后，给所有订阅者把数据推送过去。

        // publisher.subscribe(subscriber);


        for (int i = 0; i < 10; i++) {
            // if (i>=9) {
            //     publisher.closeExceptionally(new RuntimeException("数字大于9，不干了"));
            // }else {
            //     publisher.submit("p-" + i);
            // }
            System.out.println(Thread.currentThread().getName()+"\t content");
            publisher.submit("p-" + i);

            // publisher发布的所有数据在它的buffer区;
        }

        //ReactiveStream
        //jvm/底层对于整个发布订阅关系做好了异步+缓存区处理=响应式系统;
        publisher.close();


        //@formatter:off  线程暂停一会儿
        // try { TimeUnit.SECONDS.sleep( 10 ); } catch (InterruptedException e) { e.printStackTrace();}
        //@formatter:on


        //等待上面的结果全部执行完
        //数字2  main线程+守护线程  大于2说明有其他的线程
        while (Thread.activeCount() > 2) {
            //作用是让步，让当前线程由“运行状态”进入到“就绪状态”，从而让其它具有相同优先级的等待线程获取执行权
            Thread.yield();
        }
        // 发布者有数据，订阅者就会拿到
    }
}
